package com.atguigu.gmall.realtime.app.func;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.gmall.realtime.beans.TableProcess;
import org.apache.flink.api.common.state.BroadcastState;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.api.common.state.ReadOnlyBroadcastState;
import org.apache.flink.streaming.api.functions.co.BroadcastProcessFunction;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

/**
 * Author: Felix
 * Date: 2021/8/4
 * Desc: 动态分流实现
 */
public class TableProcessFunction01 extends BroadcastProcessFunction<JSONObject,String,JSONObject> {
    //声明维度侧输出流标签
    private OutputTag<JSONObject> dimTag;
    //声明广播状态描述器
    private MapStateDescriptor<String, TableProcess> mapStateDescriptor;

    //配置表构造函数
    public TableProcessFunction01(OutputTag<JSONObject> dimTag, MapStateDescriptor<String, TableProcess> mapStateDescriptor) {
        this.dimTag = dimTag;
        this.mapStateDescriptor = mapStateDescriptor;
    }

    //处理业务流中数据  maxwell从业务数据库中采集到的数据
    @Override
    public void processElement(JSONObject jsonObj, ReadOnlyContext ctx, Collector<JSONObject> out) throws Exception {
        //获取表
        String table = jsonObj.getString("table");
        //获取类型
        String type = jsonObj.getString("type");
        //注意：我们在使用Maxwell处理历史数据的时候，类型是bootstrap-insert  我们这里修复为insert
        if (type.equals("bootstrap-insert")){
            type = "insert";
            jsonObj.put("type",type);
        }
        //拼接key
        String key = table + ":" + type;
        //获取状态
        ReadOnlyBroadcastState<String, TableProcess> broadcastState = ctx.getBroadcastState(mapStateDescriptor);
        //从状态中获取配置信息
        TableProcess tableProcess = broadcastState.get(key);
        //判断在配置表中有没有该操作对应的配置
        if (tableProcess != null) {
            //在配置表中找到了该操作对应的配置  判断是事实数据还是维度数据
            String sinkTable = tableProcess.getSinkTable();
            jsonObj.put("sink_table", sinkTable);
            //获取在配置表的类型
            String sinkType = tableProcess.getSinkType();
            //判断是事实数据还是维度数据
            if (sinkType.equals(TableProcess.SINK_TYPE_HBASE)){
                //维度数据   放到维度侧输出流汇总
                ctx.output(dimTag,jsonObj);
            }else if (sinkType.equals(TableProcess.SINK_TYPE_KAFKA)){
                //事实数据  放到主流中
                out.collect(jsonObj);
            }
        }else{
            //在配置表中没有该操作对应的配置
            System.out.println("No this Key in TableProcess:"+key);
        }
    }

    //处理广播流中数据      FlinkCDC从MySQL中读取配置信息
    @Override
    public void processBroadcastElement(String jsonStr, Context ctx, Collector<JSONObject> out) throws Exception {
        //获取状态
        BroadcastState<String, TableProcess> broadcastState = ctx.getBroadcastState(mapStateDescriptor);
        //将json格式字符串转换为JSON对象
        JSONObject jsonObj = JSON.parseObject(jsonStr);
        //获取配置表中一条配置信息
        TableProcess tableProcess = JSON.parseObject(jsonObj.getString("data"), TableProcess.class);
        //业务数据库表名
        String sourceTable = tableProcess.getSourceTable();
        //操作类型
        String operateType = tableProcess.getOperateType();
        //数据类型    hbase--维度     kafka---事实
        String sinkType = tableProcess.getSinkType();
        //指定输出目的地
        String sinkTable = tableProcess.getSinkTable();
        //主键
        String sinkPk = tableProcess.getSinkPk();
        //指定保留字段
        String sinkColumns = tableProcess.getSinkColumns();
        //指定建表扩展语句
        String sinkExtend = tableProcess.getSinkExtend();
        //如果说 读取到的配置信息是维度配置的话，那么提前将维度表创建出来
        if (sinkType.equals(TableProcess.SINK_TYPE_HBASE)&&"insert".equals(operateType)){
            checkTable(sinkTable,sinkPk,sinkColumns,sinkExtend);
        }

        //拼接key
        String key = sourceTable + ":" + operateType;
        //将配置信息放到状态中
        broadcastState.put(key,tableProcess);
    }

    //在处理配置数据的时候    提前建立维度表  create table if not exists 表空间.表名(字段名 数据类型);
    private void checkTable(String sinkTable, String sinkPk, String sinkColumns, String sinkExtend) {
    }
}
